package gd1;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

class Dwd_Sql {
    public static void main(String[] args) {
        // 设置流运行环境
        EnvironmentSettings setting = EnvironmentSettings.newInstance().inStreamingMode().build();
        TableEnvironment tabEnv = TableEnvironment.create(setting);
        registerKafKaTable(tabEnv);
        Table resultTable = executeJoinQuery(tabEnv);
        resultTable.execute().print();
        sinkToKafka(tabEnv, resultTable);
    }

    private static void sinkToKafka(TableEnvironment tabEnv, Table resultTable) {
        // 实现写入Kafka的逻辑
    }

    private static Table executeJoinQuery(TableEnvironment tabEnv) {
        // 这里可以完善JOIN逻辑，示例仅返回单表
        String sql = "SELECT * FROM fact_order_detail " +
                "JOIN dim_shop_base ON fact_order_detail.shop_id = dim_shop_base.shop_id " +
                "JOIN dim_user_base ON fact_order_detail.user_id = dim_user_base.user_id";
        return tabEnv.sqlQuery(sql);
    }

    private static void registerKafKaTable(TableEnvironment tabEnv) {
        // 商品维度表
        String dim_merchandise_base = "CREATE TABLE dim_merchandise_base (\n" +
                "  merchandise_id bigint,\n" +
                "  merchandise_name String,\n" +
                "  sku_id bigint,\n" +
                "  shop_id bigint,\n" +
                "  merchandise_status String  -- 最后一列无逗号\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'dwd_dim_merchandise_base',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'json'\n" +
                ")";
        tabEnv.executeSql(dim_merchandise_base);

        // 店铺维度表
        String dim_shop_base = "CREATE TABLE dim_shop_base (\n" +
                "  shop_id bigint,\n" +
                "  shop_name string,\n" +
                "  shop_status string  -- 最后一列无逗号\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'dwd_dim_shop_base',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'json'\n" +
                ")";
        tabEnv.executeSql(dim_shop_base);

        // 用户维度表（修复核心错误：移除最后一列后的逗号）
        String dim_user_base = "CREATE TABLE dim_user_base (\n" +
                "  `user_id` bigint,\n" +
                "  `user_status` String  -- 移除了多余的逗号\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'user_behavior',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'json'\n" +
                ")";
        tabEnv.executeSql(dim_user_base);

        // 订单事实表
        String fact_order_detail = "CREATE TABLE fact_order_detail (\n" +
                "  `order_detail_id` bigint,\n" +
                "  `order_id` bigint,\n" +
                "  `merchandise_id` bigint,\n" +
                "  `sku_id` bigint,\n" +
                "  `merchandise_amount` double  -- 最后一列无逗号\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'dwd_fact_order_detail',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'json'\n" +
                ")";
        tabEnv.executeSql(fact_order_detail);
    }
}
